package me.chendan.streaming

import java.util.Properties

import kafka.javaapi.producer.Producer
import kafka.producer.{KeyedMessage, ProducerConfig}
import org.codehaus.jettison.json.JSONObject

import scala.util.Random

/**
  * Created by chendan on 16-5-4.
  */
class UserConsumeDataProducer {

}

object UserConsumeDataProducer extends App {

  private val names = Array("A", "B", "C", "D")
  private val random = new Random()

  def randomPay(): Double = {
    random.nextDouble() * 10
  }

  def getUserName(): String = {
    names(random.nextInt(names.length))
  }

  val topic = "UserConsume"
  val broker = "localhost:9092"
  val props = new Properties()
  props.put("metadata.broker.list", broker)
  props.put("serializer.class", "kafka.serializer.StringEncoder")
  val kafkaConfig = new ProducerConfig(props)
  val producer = new Producer[String, String](kafkaConfig)

  // {"user":"A","event_time":"1430204612405","payment":"9.0"}
  while(true) {
    // 创建json串
    val event = new JSONObject()
    event
      .put("user", getUserName())
      .put("payment", randomPay())
      .put("even_time", System.currentTimeMillis())

    // 往kafka发送数据
    producer.send(new KeyedMessage[String, String](topic, event.toString))
    println("Message sent: " + event)

    //每隔200ms发送一条数据
    Thread.sleep(500)
  }
}
